package com.fwmagic.flink.state.source;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.RandomAccessFile;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

/**
 * 多并行source需继承RichParallelSourceFunction
 * 自己维护状态（这里是OperatorState）到checkpoint需实现CheckpointedFunction
 */
public class MyExcetlyOnceParallFileSource extends RichParallelSourceFunction<Tuple2<Integer, String>> implements CheckpointedFunction {

    //transient不参与序列化
    private transient ListState<Long> listState;

    //控制循环
    private Boolean flag = true;

    private Long offset = 0L;

    //读取文件的目录：/Users/fa ngwei/temp/data
    private String path;

    public MyExcetlyOnceParallFileSource() {
    }

    public MyExcetlyOnceParallFileSource(String path) {
        this.path = path;
    }

    /**
     * 程序运行一次的方法，要想一直度文件，可以用while循环读
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void run(SourceContext ctx) throws Exception {
        //获取历史listState中的offset,为了任务重启时不重复读取数据
        Iterator<Long> iterator = listState.get().iterator();
        while(iterator.hasNext()){
            offset = iterator.next();
        }

        //获取subTaks的索引
        int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();

        //因为是并发执行，会有线程不安全的文件，需要在更新offset和collect数据时加锁
        Object checkpointLock = ctx.getCheckpointLock();

        RandomAccessFile raf = new RandomAccessFile(path + "/" + indexOfThisSubtask + ".txt", "r");
        //从指定位置读取文件
        raf.seek(offset);
        //读文件
        while (flag) {
            String line = raf.readLine();
            if (line != null) {
                line = new String(line.getBytes("ISO-8859-1"), "utf-8");
                synchronized (checkpointLock) {
                    ctx.collect(Tuple2.of(indexOfThisSubtask, line));
                    //每次读取文件内容，获取读取到的指针，更新offset
                    offset = raf.getFilePointer();
                    System.out.println("====>当前offset:"+offset);
                }
            } else {
                //没有数据时，休眠1s
                TimeUnit.SECONDS.sleep(1);
            }
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }

    /**
     * 会调用多次，每隔固定的时间就会做checkpoint操作，生成快照，
     * 将数据保存到StateBackend中
     *
     * @param context
     * @throws Exception
     */
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        //清空状态
        listState.clear();
        //更新状态
        listState.add(offset);

        //获取CheckPoint执行的时间和次数
        long checkpointId = context.getCheckpointId();
        long checkpointTimestamp = context.getCheckpointTimestamp();
        String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(checkpointTimestamp));
        System.err.println("=====CheckPoint执行,生成快照。时间：" + date + ",第" + checkpointId + "次执行!");
    }

    /**
     * 仅会调用一次
     * 用户初始化OperatorState
     *
     * @param context
     * @throws Exception
     */
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        OperatorStateStore operatorStateStore = context.getOperatorStateStore();

        ListStateDescriptor<Long> listStateDescriptor = new ListStateDescriptor<>("op-state-1", Long.class);

        listState = operatorStateStore.getListState(listStateDescriptor);
    }
}
